package com.soup.lab.basic.iso;

import com.segi.uhomecp.park.access.util.empty.EmptyUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * 用于实现任务列表按 Key 线程隔离
 *
 * @param <K>   任务分类键的类型
 * @param <V>   任务标识
 * @author Soup
 */
@Slf4j
public class IsolateTaskScheduler<K, V> {

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>());

    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private final Object startupMonitor = new Object();

    private final ConcurrentMap<K, BlockingDeque<V>> dequeRegistry;

    private final TaskSource<K, V> taskSource;

    private final Consumer<V> task;

    private volatile boolean started;
    /**
     * 是否允许线程在没有任务的情况下，从其他键的任务队列的队列尾偷取任务执行
     * 如果任务必须顺序执行，应设置为 false
     */
    private volatile boolean steal = true;

    public IsolateTaskScheduler(ConcurrentMap<K, BlockingDeque<V>> dequeRegistry, TaskSource<K, V> taskSource, Consumer<V> task) {
        this.dequeRegistry = dequeRegistry;
        this.taskSource = taskSource;
        this.task = task;
    }

    public IsolateTaskScheduler(TaskSource<K, V> taskSource, Consumer<V> execution) {
        this.dequeRegistry = new ConcurrentHashMap<>();
        this.taskSource = taskSource;
        this.task = execution;
    }

    public void start() {
        if(!started) {
            synchronized (startupMonitor) {
                if(!started) {
                    scheduledExecutorService.scheduleAtFixedRate(new KeyRegisterTask(), 0, 30, TimeUnit.SECONDS);
                    started = true;
                }
            }
        }
    }

    public void stop() {
        if(started) {
            synchronized (startupMonitor) {
                if (started) {
                    dequeRegistry.clear();
                    executor.shutdown();
                    scheduledExecutorService.shutdown();
                    started = false;
                }
            }
        }
    }

    public void setSteal(boolean steal) {
        this.steal = steal;
    }

    public boolean isSteal() {
        return steal;
    }

    public boolean isStarted() {
        return started;
    }

    /**
     * 获取所有任务key,确保每一个key都有一个队列
     */
    class KeyRegisterTask implements Runnable {

        @Override
        public void run() {
            List<K> keys = taskSource.keys();
            for (K key : keys) {
                if(dequeRegistry.putIfAbsent(key, new LinkedBlockingDeque<>()) == null) {
                    log.info("创建了任务：{}", key);
                    executor.submit(new SelfishQueuedTask(key));
                }
            }
            if(keys.size() > executor.getCorePoolSize()) {
                executor.setMaximumPoolSize(keys.size());
                executor.setCorePoolSize(keys.size());
                executor.allowCoreThreadTimeOut(false);
            } else if(keys.size() < executor.getCorePoolSize() / 2) {
                // 当键的数量小于 线程池核心线程数量的一半时，允许回收空闲的核心线程
                executor.allowCoreThreadTimeOut(true);
            }
        }

    }

    /**
     * “自私”的任务
     * 有的时候，它是一个自私的任务，只会执行和自己key相关的事情。
     * 有的时候，它又是一个热心肠的任务，会偷偷帮助其他人，不留姓名。
     * 真是让人摸不透呢。
     */
    class SelfishQueuedTask implements Runnable {
        final K key;

        SelfishQueuedTask(K key) {
            this.key = key;
        }

        @Override
        public void run() {
            V v = null;
            while(true) {
                try {
                    if ((v = next(key)) == null) {
                        break;
                    }
                    task.accept(v);
                } catch (Exception e) {
                    log.error("执行任务异常：{}", v);
                }
            }
            log.info("任务处理完了", key);
        }

        private V next(K key) {
            BlockingDeque<V> blockingDeque = dequeRegistry.get(key);
            if(blockingDeque == null) {
                return null;
            }
            V v = blockingDeque.pollFirst();
            if(v != null) {
                return v;
            }
            List<V> values = taskSource.fetchByKey(key);
            log.info("获取任务：{}，结果：{}", key, values);
            if(EmptyUtils.isNotEmpty(values)) {
                values.forEach(blockingDeque::offer);
                return blockingDeque.pollFirst();
            } else if(steal) {
                // steal 从其他任务队列尾中偷一个
                for (BlockingDeque<V> q : dequeRegistry.values()) {
                    if(q != blockingDeque && (v = q.pollLast()) != null) {
                        log.info("{} steal: {}", key, v);
                        return v;
                    }
                }
            }
            log.info("移除了队列：{}", key);
            dequeRegistry.remove(key);
            return null;
        }
    }

}
